package com.example.testpulsar.mq.configforpulsar;

import org.apache.pulsar.client.api.*;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.TimeUnit;

/**
 * @description: pulsar配置类
 * @author LiuZihan
 * @date 2021/8/16 11:02
 * @version 1.0
 */
@Configuration
@ConditionalOnProperty(prefix = "jms.pulsar" ,value ="enable",havingValue = "true")
public class PulsarConfiguration {

    /**
     * @description: 创建PulsarClient对象
     * @param: environment
     * @return: org.apache.pulsar.client.api.PulsarClient
     * @author LiuZihan
     * @date: 2021/8/16 15:15
     */
    @Bean
    public PulsarClient createPulsarConnection(PulsarProperties pulsarProperties){
        String url = pulsarProperties.getUrl();
        String operationTimeoutStr = pulsarProperties.getOperationTimeout();
        int operationTimeout = Integer.parseInt(operationTimeoutStr);
        String numIoThreadsStr = pulsarProperties.getNumIoThreads();
        int numIoThreads = Integer.parseInt(numIoThreadsStr);
        String keepAliveIntervalStr = pulsarProperties.getKeepAliveInterval();
        int keepAliveInterval = Integer.parseInt(keepAliveIntervalStr);
        String durationStr = pulsarProperties.getDuration();
        int duration = Integer.parseInt(durationStr);

        PulsarClient client = null;
        //构造Pulsar client
        try {
            client = PulsarClient.builder()
                    .serviceUrl(url)
                    .operationTimeout(operationTimeout, TimeUnit.MILLISECONDS) //操作超时时长，默认30000ms
                    .ioThreads(numIoThreads) //用于处理到brokers的连接的线程数
                    .listenerThreads(1) //用于处理消息监听器的线程数。如果您希望多个线程处理单个主题，则需要创建一个shared订阅，并为此订阅创建多个消费者。这并不能确保排序。
                    .maxConcurrentLookupRequests(5000) //允许在每个broker连接上发送的并发查找请求数，以防止broker过载
                    .maxLookupRequests(50000) //每个broker连接上允许的最大查找请求数，以防止broker过载
                    .keepAliveInterval(keepAliveInterval, TimeUnit.SECONDS) //每个客户端broker连接的保持活跃间隔的秒数，默认30s
                    .connectionTimeout(duration, TimeUnit.MILLISECONDS) //等待建立连接到broker的持续时间，如果持续时间过了，没有从broker得到响应，连接尝试将被放弃
                    .build();
        } catch (PulsarClientException e) {
            //TODO 做日志记录，创建Pulsar连接失败
        }
        return client;
    }

    public static Producer<String> getProducer(PulsarClient client, PulsarProperties pulsarProperties,
                                               boolean persistent, String tenant, String nameSpace, String topic) throws PulsarClientException {
        String persistentMode = "persistent";
        if(!persistent){
            persistentMode = "non-persistent";
        }
        String topicName = persistentMode + "://" + tenant + "/" + nameSpace + "/" + topic;

        String enableBatchingStr = pulsarProperties.getEnableBatching();
        boolean enableBatching = Boolean.parseBoolean(enableBatchingStr);
        String sendTimeoutStr = pulsarProperties.getSendTimeout();
        int sendTimeout = Integer.parseInt(sendTimeoutStr);
        String blockIfQueueFullStr = pulsarProperties.getBlockIfQueueFull();
        boolean blockIfQueueFull = Boolean.parseBoolean(blockIfQueueFullStr);
        String compressionTypeStr = pulsarProperties.getCompressionType();
        CompressionType compressionType;
        switch (compressionTypeStr){
            case "1":
                compressionType = CompressionType.LZ4;
                break;
            case "2":
                compressionType = CompressionType.ZLIB;
                break;
            case "3":
                compressionType = CompressionType.ZSTD;
                break;
            case "4":
                compressionType = CompressionType.SNAPPY;
                break;
            case "0":
            default:
                compressionType = CompressionType.NONE;
                break;
        }

        String batcherBuilderStr = pulsarProperties.getBatcherBuilder();
        BatcherBuilder batcherBuilder;
        switch (batcherBuilderStr){
            case "1":
                batcherBuilder = BatcherBuilder.KEY_BASED;
                break;
            case "0":
            default:
                batcherBuilder = BatcherBuilder.DEFAULT;
                break;
        }

        //创建producer
        Producer<String> producer = client.newProducer(Schema.STRING)
                .topic(topicName)
                .enableBatching(enableBatching)//是否开启批量处理消息，默认true,需要注意的是enableBatching只在异步发送sendAsync生效，同步发送send失效。因此建议生产环境若想使用批处理，则需使用异步发送，或者多线程同步发送
                .compressionType(compressionType)//消息压缩（四种压缩方式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不用做改动就能消费，开启后大约可以降低3/4带宽消耗和存储（官方测试）
                .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) //设置将对发送的消息进行批处理的时间段,10ms；可以理解为若该时间段内批处理成功，则一个batch中的消息数量不会被该参数所影响。
                .sendTimeout(sendTimeout, TimeUnit.SECONDS)//设置发送超时0s；如果在sendTimeout过期之前服务器没有确认消息，则会发生错误。默认30s，设置为0代表无限制，建议配置为0
                .batchingMaxMessages(1000)//批处理中允许的最大消息数。默认1000
                .maxPendingMessages(1000)//设置等待接受来自broker确认消息的队列的最大大小，默认1000
                .blockIfQueueFull(blockIfQueueFull)//设置当消息队列中等待的消息已满时，Producer.send 和 Producer.sendAsync 是否应该block阻塞。默认为false，达到maxPendingMessages后send操作会报错，设置为true后，send操作阻塞但是不报错。建议设置为true
                .roundRobinRouterBatchingPartitionSwitchFrequency(10)//向不同partition分发消息的切换频率，默认10ms，可根据batch情况灵活调整
                .batcherBuilder(batcherBuilder)//key_Shared模式要用KEY_BASED,才能保证同一个key的message在一个batch里
                .create();
        return producer;
    }

    public static Consumer<String> getConsumer(PulsarClient client, PulsarProperties pulsarProperties,
                                               MessageListener<String> messageListener,
                                               boolean persistent, String tenant, String nameSpace, String topic,
                                               String subscription, SubscriptionType subscriptionType) throws PulsarClientException {
        String persistentMode = "persistent";
        if(!persistent){
            persistentMode = "non-persistent";
        }
        String topicName = persistentMode + "://" + tenant + "/" + nameSpace + "/" + topic;

        String ackTimeoutStr = pulsarProperties.getAckTimeout();
        long ackTimeout = Long.parseLong(ackTimeoutStr);
        String subscriptionInitialPositionStr = pulsarProperties.getSubscriptionInitialPosition();
        SubscriptionInitialPosition subscriptionInitialPosition;
        switch (subscriptionInitialPositionStr){
            case "1":
                subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
                break;
            case "0":
            default:
                subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
                break;
        }
        String redeliveryDelayStr = pulsarProperties.getRedeliveryDelay();
        long redeliveryDelay = Long.parseLong(redeliveryDelayStr);

        //创建consumer
        Consumer<String> consumer = client.newConsumer(Schema.STRING)
                .topic(topicName)
                .messageListener(messageListener)
                .subscriptionName(subscription)
                .subscriptionType(subscriptionType)
                .ackTimeout(ackTimeout, TimeUnit.MILLISECONDS) //未确认消息的超时时间，默认值 0
                .subscriptionInitialPosition(subscriptionInitialPosition)//指定从哪里开始消费还有Latest，valueof可选，默认Latest
                .negativeAckRedeliveryDelay(redeliveryDelay, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .subscribe();
        return consumer;
    }
}
